package com.shujia.spark.stream

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object Demo3Window {
  def main(args: Array[String]): Unit = {
    //创建spark环境
    val conf = new SparkConf()
    conf.setAppName("state")
    conf.setMaster("local[2]")
    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    ssc.checkpoint("data/checkpoint")

    //读取数据
    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    //转换成kv格式
    val kvDS: DStream[(String, Int)] = linesDS
      .flatMap(line => line.split(","))
      .map(word => (word, 1))

    /**
     * reduceByKeyAndWindow" 窗口计算，每隔一段时间计算最近一段时间的数据
     */

    /*    val countDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
          reduceFunc = (x, y) => x + y, //聚合函数
          windowDuration = Durations.seconds(15), //窗口大小
          slideDuration = Durations.seconds(5) //滑动时间
        )*/

    //对滑动拆给你扣的计算进行优化，可以比秒重复计算

    val countDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
      (x, y) => x + y, //加的函数
      (i, j) => i - j, //减的函数
      windowDuration = Durations.seconds(15), //窗口大小
      slideDuration = Durations.seconds(5) //滑动时间
    )

    countDS.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
